"""
完成练习案例：JSON商品统计
需求：
1、各个城市销售额排名，从大到小
2、全部城市，有哪些商品类别在售卖
3、北京市有哪些商品类型在售卖
"""
from pyspark import SparkConf, SparkContext
import os
import json

os.environ['PYSPARK_PYTHON'] = 'D:/Work/anaconda3/python.exe'

conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)

# 读取文件数据
file_rdd = sc.textFile("orders.txt")
# 取出一个个的JSON字符串
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
# 将JSON字符串转换成字典
dict_rdd = json_str_rdd.map(lambda x: json.loads(x))

# TODO 需求1：各个城市销售额排名，从大到小
# 取出城市和销售额数据
# (城市,销售额)
city_money_rdd = dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
# 按照城市分组销售额聚合
city_result = city_money_rdd.reduceByKey(lambda x, y: x + y)
# 按照销售额聚合结果进行排序(降序，从大到小)
city_result_sort = city_result.sortBy(lambda x: x[1], ascending=False)
print(f"需求1的结果是：{city_result_sort.collect()}")

# TODO 需求2：全部城市，有哪些商品类别在售卖
# 取出所有记录的商品类别字段
category_list_rdd = dict_rdd.map(lambda x: x['category'])
# 对商品类别去重
category_distinct_rdd = category_list_rdd.distinct()
print(f"需求2的结果是：{category_distinct_rdd.collect()}")

# TODO 需求3：北京市有哪些商品类型在售卖
# 对所有的订单记录进行过滤，只选择城市为北京的
beijing_order_rdd = dict_rdd.filter(lambda x: x['areaName'] == "北京")
# 取出北京的订单的商品类别字段
beijing_category_list_rdd = beijing_order_rdd.map(lambda x: x['category'])
# 对商品类别去重
beijing_category_distinct_rdd = beijing_category_list_rdd.distinct()

print(f"需求3:的结果是：{beijing_category_distinct_rdd.collect()}")
